package com.yl.flink.processor;

import cn.hutool.core.date.DateUtil;
import com.yl.constant.CdcType;
import com.yl.constant.Const;
import com.yl.entity.cdc.*;
import com.yl.util.MysqlCdcUtil;
import com.yl.util.SUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Optional;

/**
 * @author wlf
 * @since 2022/8/10
 */
@Slf4j
public class MysqlCdcParseFunc extends ProcessFunction<String, MysqlCdcEntity> {

    /**
     * 把获取的各个表的json串解析成对应的表对象
     * 【注意：】
     * 每个表对象的key，是指数据处理时如何获取表对象
     * <p>
     * 【mysql-cdc配置】
     * 1.开启 binlog
     * server_id=1
     * log_bin=mysql-bin
     * binlog_format=ROW
     * expire_logs_days=1
     * 2.赋权
     * SELECT 允许从表中查看数据
     * REPLICATION SLAVE 允许执行show master status,show slave status,show binary logs命令
     * REPLICATION CLIENT 允许slave主机通过此用户连接master以便建立主从复制关系
     *
     * @param cdcJson cdc输出json串
     * @param context 上下文环境
     * @param out     输出器
     */
    @Override
    public void processElement(String cdcJson, Context context, Collector<MysqlCdcEntity> out) throws Exception {
        // 项目信息
        Optional
                .ofNullable(MysqlCdcUtil.parse(cdcJson, CdcType.PRO.getTable(), BaseProject.class))
                .ifPresent(cdcEntity -> {
                    // 使用时如何获取对象
                    String addKey = SUtil.fmt(cdcEntity.getEntity().getId());
                    cdcEntity.setType(CdcType.PRO);
                    cdcEntity.setAddKey(addKey);
                    cdcEntity.setDelKey(addKey);
                    out.collect(cdcEntity);
                    log.info("cdc: {}", cdcEntity);
                });

        // 用户信息
        Optional
                .ofNullable(MysqlCdcUtil.parse(cdcJson, CdcType.USR.getTable(), AuthUser.class))
                .ifPresent(cdcEntity -> {
                    AuthUser entity = cdcEntity.getEntity();
                    // 判断是否接收告警，前提是用户状态正常
                    if (entity.getStatus() == 1 && !Const.S_ZERO.equals(entity.getReceive_warn())) {
                        // 使用时如何获取对象
                        String addKey = SUtil.fmt(entity.getProject_id());
                        String delKey = SUtil.fmt(entity.getProject_id(), entity.getId());
                        cdcEntity.setType(CdcType.USR);
                        cdcEntity.setAddKey(addKey);
                        cdcEntity.setDelKey(delKey);
                        out.collect(cdcEntity);
                        log.info("cdc: {}", cdcEntity);
                    }
                });

        // 测点信息
        Optional
                .ofNullable(MysqlCdcUtil.parse(cdcJson, CdcType.TGT.getTable(), BaseTarget.class))
                .ifPresent(cdcEntity -> {
                    BaseTarget entity = cdcEntity.getEntity();
                    // 使用时如何获取对象
                    String addKey = SUtil.fmt(entity.getProject_id(), entity.getSerial_number());
                    cdcEntity.setType(CdcType.TGT);
                    cdcEntity.setAddKey(addKey);
                    cdcEntity.setDelKey(addKey);
                    out.collect(cdcEntity);
                    log.info("cdc: {}", cdcEntity);
                });

        // 测点类型信息
        Optional
                .ofNullable(MysqlCdcUtil.parse(cdcJson, CdcType.TYP.getTable(), BaseTargetType.class))
                .ifPresent(cdcEntity -> {
                    // 使用时如何获取对象
                    String addKey = SUtil.fmt(cdcEntity.getEntity().getId());
                    cdcEntity.setType(CdcType.TYP);
                    cdcEntity.setAddKey(addKey);
                    cdcEntity.setDelKey(addKey);
                    out.collect(cdcEntity);
                    log.info("cdc: {}", cdcEntity);
                });

        // 测点阈值信息
        Optional
                .ofNullable(MysqlCdcUtil.parse(cdcJson, CdcType.THD.getTable(), SettingsThreshold.class))
                .ifPresent(cdcEntity -> {
                    SettingsThreshold entity = cdcEntity.getEntity();
                    // 使用时如何获取对象
                    String addKey = SUtil.fmt(entity.getTarget_id(), entity.getQuota_code());
                    String delKey = SUtil.fmt(entity.getId());
                    cdcEntity.setType(CdcType.THD);
                    cdcEntity.setAddKey(addKey);
                    cdcEntity.setDelKey(delKey);
                    out.collect(cdcEntity);
                    log.info("cdc: {}", cdcEntity);
                });

        // 指标计算信息
        Optional
                .ofNullable(MysqlCdcUtil.parse(cdcJson, CdcType.QTA.getTable(), SettingsDataQuota.class))
                .ifPresent(cdcEntity -> {
                    SettingsDataQuota entity = cdcEntity.getEntity();
                    // 判断是否是启用指标
                    if (entity.getEnable().equals(1)) {
                        // 使用时如何获取对象
                        String addKey = SUtil.fmt(entity.getTarget_type_id(), entity.getCode());
                        cdcEntity.setType(CdcType.QTA);
                        cdcEntity.setAddKey(addKey);
                        cdcEntity.setDelKey(addKey);
                        out.collect(cdcEntity);
                        log.info("cdc: {}", cdcEntity);
                    }
                });

        // 批次完整性信息（协议3）
        Optional
                .ofNullable(MysqlCdcUtil.parse(cdcJson, CdcType.BAT.getTable(), BatchIntegrity.class))
                .ifPresent(cdcEntity -> {
                    BatchIntegrity entity = cdcEntity.getEntity();
                    // 日期字段时区问题，比实际多8个小时
                    final long ts8h = 8 * 3600 * 1000;
                    Optional
                            .ofNullable(entity.getStart_time())
                            .ifPresent(t -> entity.setStart_time(DateUtil.date(Long.parseLong(t) - ts8h).toString()));
                    Optional
                            .ofNullable(entity.getEnd_time())
                            .ifPresent(t -> entity.setEnd_time(DateUtil.date(Long.parseLong(t) - ts8h).toString()));
                    Optional
                            .ofNullable(entity.getCreate_time())
                            .ifPresent(t -> entity.setCreate_time(DateUtil.date(Long.parseLong(t) - ts8h).toString()));
                    // 使用时如何获取对象
                    String theDay = DateUtil.parse(entity.getStart_time()).toString(Const.FMT_TRIM_DAY);
                    String addKey = SUtil.fmt(entity.getTarget_id(), theDay, entity.getBatch_id());
                    cdcEntity.setType(CdcType.BAT);
                    cdcEntity.setAddKey(addKey);
                    cdcEntity.setDelKey(addKey);
                    out.collect(cdcEntity);
                    log.info("cdc: {}", cdcEntity);
                });
    }

}
